import { AsyncSubject, EMPTY, merge, Observable, of, Subject } from "rxjs"
import { catchError, concatWith, distinctUntilChanged, ignoreElements, last, map, mapTo, mergeMap, scan, share, shareReplay, take, takeUntil, tap, withLatestFrom } from "rxjs/operators"
import { xtp } from "./pb";

//XTP连接，底层可使用任意有序的传输协议（例如TCP, WebSocket, QUIC）等
export interface Connection {
    //收到的数据帧
    readonly onFrame: Observable<xtp.Frame>,
    //数据帧发送器
    readonly frameSender: Subject<xtp.IFrame>,
}

//XTP端（客户端或服务端）
export interface Peer {
    //收到对端的根处理者
    readonly onRootRemoteHandler: Observable<RemoteHandler>
    //创建处理者
    readonly createHandler: (handlerInfo: xtp.HandlerInfo) => Handler
    //声明根处理者信息
    readonly declareRootHandlerInfo: (handler: Handler) => void
    //关闭连接
    readonly close: () => void
}

//处理者
export interface Handler {
    //处理者信息
    readonly info: xtp.IHandlerInfo
    //收到流消息
    readonly onData: Observable<Uint8Array>
    //接收到对端处理者信息
    readonly onRemoteHandler: Observable<RemoteHandler>
    //产出消息(Message.data)的发送器
    readonly yieldedDataSender: Subject<Uint8Array>
    //处理者生命周期结束
    readonly theEnd: Observable<null>
    //产出HandlerInfo消息
    readonly yieldHandler: (handler: Handler) => void
}

//远端处理者
export interface RemoteHandler {
    //处理者信息
    readonly info: xtp.IHandlerInfo
    //消息(Message.data)的发送器
    readonly dataSender: Subject<Uint8Array>
    //是否可发送消息
    readonly onAvailable: Observable<boolean>
    //收到的产出消息
    readonly onYieldedData: Observable<Uint8Array>
    //接收到对端产出的处理者信息
    readonly onRemoteHandler: Observable<RemoteHandler>
    //处理者生命周期结束
    readonly theEnd: Observable<null>
    //发送HandlerInfo消息
    readonly sendHandlerInfo: (info: xtp.HandlerInfo) => void
}

//私有的上下文
interface Context {
    readonly conn: Connection
    readonly watchMessageFrames: (flowId: number) => Observable<xtp.Frame>
    readonly watchCloseFrames: (flowId: number) => Observable<xtp.Frame>
    readonly watchYieldFrames: (flowId: number) => Observable<xtp.Frame>
    readonly watchEndFrames: (flowId: number) => Observable<xtp.Frame>
    readonly newHandlerId: () => number
}

//比groupBy+find更有效率
export const getSubValues = <V, K>(values: Observable<V>, keySelector: (value: V) => K): (key: K) => Observable<V> => {
    const m = new Map<K, Subject<V>>()
    const theEnd = values.pipe(
        tap(v => {
            const key = keySelector(v)
            const subject = m.get(key)
            if (subject)
                subject.next(v)
        }),
        last(null, 1),
        share(),
    )
    return key => new Observable<V>(s => {
        let subject = m.get(key)
        if (!subject) {
            subject = new Subject<V>()
            m.set(key, subject)
        }
        const sub = subject.subscribe(s)
        return () => {
            sub.unsubscribe()
            if (subject.observers.length == 0) m.delete(key)
        }
    }).pipe(
        takeUntil(theEnd),
    )
}
// export const getSubValues = <V, K>(values: Observable<V>, keySelector: (value: V) => K): (key: K) => Observable<V> => {
//     const m = new Map<K, Set<Observer<V>>>()
//     const theEnd = values.pipe(
//         tap(v => {
//             const observerSet = m.get(keySelector(v))
//             if (!observerSet) return
//             observerSet.forEach(observer => observer.next(v))
//         }),
//         last(null, 1),
//         share(),
//     )
//     return key => new Observable<V>(s => {
//         let observerSet = m.get(key)
//         if (!observerSet) {
//             observerSet = new Set<Observer<V>>()
//             m.set(key, observerSet)
//         }
//         observerSet.add(s)
//         return () => {
//             observerSet.delete(s)
//             if (observerSet.size == 0) m.delete(key)
//         }
//     }).pipe(
//         takeUntil(theEnd),
//     )
// }

//Uint8Array转string
export const getString = (u8arr: Uint8Array) => new Observable<string>(s => {
    const b = new Blob([u8arr])
    const fr = new FileReader()
    fr.readAsText(b, 'utf-8')
    fr.onerror = function (ev) {
        s.error('bad Uint8Array')
    }
    fr.onloadend = function () {
        s.next(fr.result as string)
        s.complete()
    }
})

//string转Uint8Array
export const getU8Array = (str: string) => new Observable<Uint8Array>(s => {
    if (typeof str != 'string') return s.error('expect a string')
    const b = new Blob([str])
    const fr = new FileReader()
    fr.readAsArrayBuffer(b)
    fr.onloadend = function () {
        const buf = new Uint8Array(fr.result as Uint8Array)
        s.next(buf)
        s.complete()
    }
})

//从H5的WebSocket创建XTP连接
export const fromH5WebSocket = (url: string) => new Observable<Connection>(s => {
    const ws = new WebSocket(url, 'xtp')
    ws.binaryType = "arraybuffer"
    ws.onclose = (ev) => {
        s.error(ev.reason)
    }
    ws.onopen = (ev) => {
        ws.onclose = null
        const theEnd = new Observable<void>(s2 => {
            const closeCb = (ev: CloseEvent) => {
                s2.error(ev.reason)
            }
            ws.addEventListener('close', closeCb)
            return () => {
                ws.removeEventListener('close', closeCb)
            }
        }).pipe(
            share(),
        )
        const onFrame = new Observable<xtp.Frame>(s2 => {
            const cb = (ev: MessageEvent) => {
                const buf = new Uint8Array(ev.data)
                // console.info("received:", buf)
                const frame = xtp.Frame.decode(buf)
                s2.next(frame)
            }
            ws.addEventListener("message", cb)
            return () => {
                ws.removeEventListener("message", cb)
            }
        }).pipe(
            takeUntil(theEnd),
            share(),
        )
        const frameSender = new Subject<xtp.IFrame>()
        //side effect
        frameSender.pipe(
            takeUntil(theEnd),
        ).subscribe({
            next:frame => {
                // console.info("send:", frame)
                const buf = xtp.Frame.encode(frame).finish()
                ws.send(buf)
            },
            error:err => {
                // console.log("frameSender onError")
                ws.close(4000, err.toString())
            },
            complete:() => {
                // console.log("frameSender onComplete")
                ws.close()
            }
        })
        s.next({
            onFrame,
            frameSender,
        })
        s.complete()
    }
})

//把xtp连接转换为Peer
export const toPeer = (conn: Connection): Peer => {
    const ctx = newContext(conn)
    return newPeer(ctx)
}

const newProtocolError = (message: string): Error => ({ name: 'ProtocolError', message })

const newRemoteError = (name: string, message: string): Error => ({ name, message })

const newContext = (conn: Connection): Context => {
    const { onFrame } = conn
    const getFramesByType = getSubValues(onFrame, frame => frame.type)
    const messageFrames = getFramesByType('message')
    const closeFrames = getFramesByType('close')
    const yieldFrames = getFramesByType('yield')
    const endFrames = getFramesByType('end')
    const watchMessageFrames = getSubValues(messageFrames, frame => frame.handlerId)
    const watchCloseFrames = getSubValues(closeFrames, frame => frame.handlerId)
    const watchYieldFrames = getSubValues(yieldFrames, frame => frame.handlerId)
    const watchEndFrames = getSubValues(endFrames, frame => frame.handlerId)
    let handlerId = 1
    const newHandlerId = () => handlerId++
    return {
        conn,
        watchMessageFrames,
        watchCloseFrames,
        watchYieldFrames,
        watchEndFrames,
        newHandlerId,
    }
}

const newPeer = (ctx: Context): Peer => {
    const frameSender = ctx.conn.frameSender
    const onMessageFrame = ctx.watchMessageFrames(0)
    const firstRemoteHandler = onMessageFrame.pipe(
        map(frame => {
            const message = frame.message
            if (!message.handlerInfo)
                throw newProtocolError("first message type must be HANDLER_INFO")
            const info = message.handlerInfo
            if (info.handlerId <= 0)
                throw newProtocolError("first handlerId must greater than 0")
            return newRemoteHandler(ctx, info)
        }),
        take(1),
        shareReplay(),
    )
    const rootHandlerInfoSender = new AsyncSubject<Handler>()
    const singleRootHandler = rootHandlerInfoSender.pipe(
        tap(handler => {
            //发送handlerInfo
            frameSender.next({
                handlerId: 0,
                message: {
                    handlerInfo: handler.info,
                }
            })
        }),
        shareReplay(),
    )
    //side effect
    singleRootHandler.subscribe()
    firstRemoteHandler.subscribe()
    onMessageFrame.subscribe()
    return {
        onRootRemoteHandler: firstRemoteHandler,
        createHandler: handlerInfo => newHandler(ctx, handlerInfo),
        declareRootHandlerInfo: handler => {
            rootHandlerInfoSender.next(handler)
            rootHandlerInfoSender.complete()
        },
        close: () => ctx.conn.frameSender.complete(),
    }
}

const newRemoteHandler = (ctx: Context, handlerInfo: xtp.IHandlerInfo): RemoteHandler => {
    const frameSender = ctx.conn.frameSender
    const handlerId = handlerInfo.handlerId
    const onEnd = ctx.watchEndFrames(handlerId).pipe(
        take(1),
        mergeMap(frame => {
            const error = frame.end.error
            if (error) {
                throw newRemoteError(error.typeName, error.textMessage)
            }
            // return EMPTY
            return of(null)
        }),
        shareReplay(),
    )
    const onYield = ctx.watchYieldFrames(handlerId).pipe(
        map(frame => frame.yield as xtp.Message),
        takeUntil(onEnd),
        share(),
    )
    const watchYieldByType = getSubValues(onYield, msg => msg.type)
    const onRemoteHandler = watchYieldByType("handlerInfo").pipe(
        map(msg => newRemoteHandler(ctx, msg.handlerInfo)),
        share(),
    )
    const onYieldedData = watchYieldByType("data").pipe(
        map(msg => msg.data),
        share(),
    )
    const dataSender = new Subject<Uint8Array>()
    const handlerInfoSender = new Subject<xtp.IHandlerInfo>()
    const availableDataSender = new Subject<Uint8Array>()
    const onSentData = availableDataSender.pipe(
        tap({
            next: data => frameSender.next({
                handlerId,
                message: { data }
            }),
            complete: () => frameSender.next({
                handlerId,
                end: {}
            }),
            error: err => frameSender.next({
                handlerId,
                end: {
                    error: { typeName: err.name || 'Error', textMessage: err.message }
                }
            }),
        }),
        catchError(err => EMPTY),
        share(),
    )
    const availableHandlerSender = new Subject<xtp.IHandlerInfo>()
    const onSentHandlerInfo = availableHandlerSender.pipe(
        tap({
            next: handlerInfo => frameSender.next({
                handlerId,
                message: { handlerInfo }
            })
        }),
        share(),
    )
    const onSentMessage = merge(
        onSentHandlerInfo,
        onSentData
    )
    const onAvailableAmount = merge(
        of(handlerInfo.channelSize),
        onSentMessage.pipe(mapTo(-1)),
        onYieldedData.pipe(mapTo(1)),
        onRemoteHandler.pipe(
            mergeMap(it => it.theEnd),
            mapTo(1),
        ),
    ).pipe(
        scan((a, b) => a + b, 0),
        shareReplay(1),
    )
    const onAvailable = onAvailableAmount.pipe(
        map(amount => amount > 0),
        distinctUntilChanged(),
        shareReplay(1),
    )
    //side effect
    dataSender.pipe(
        withLatestFrom(onAvailable),
        tap({
            next: ([data, ok]) => {
                if (ok) availableDataSender.next(data)
            },
            complete: availableDataSender.complete,
            error: availableDataSender.error,
        }),
        catchError(err => EMPTY),
    ).subscribe()
    handlerInfoSender.pipe(
        withLatestFrom(onAvailable),
        tap({
            next: ([handler, ok]) => {
                if (ok) availableHandlerSender.next(handler)
            },
        }),
        catchError(err => EMPTY),
    ).subscribe()
    return {
        info: handlerInfo,
        dataSender,
        onAvailable,
        onYieldedData,
        onRemoteHandler,
        theEnd: onEnd,
        sendHandlerInfo: it=>handlerInfoSender.next(it),
    }
}

const newHandler = (ctx: Context, handlerInfo: xtp.IHandlerInfo): Handler => {
    const frameSender = ctx.conn.frameSender
    const handlerId = ctx.newHandlerId()
    handlerInfo.handlerId = handlerId
    const onClose = ctx.watchCloseFrames(handlerId).pipe(
        take(1),
        mergeMap(frame => {
            const error = frame.end.error
            if (error) {
                throw newRemoteError(error.typeName, error.textMessage)
            }
            // return EMPTY
            return of(null)
        }),
        shareReplay(),
    )
    const yieldedDataSender = new Subject<Uint8Array>()
    const onEnd = yieldedDataSender.pipe(
        ignoreElements(),
        concatWith(of(null)),
        shareReplay(),
    )
    const onMessage = ctx.watchMessageFrames(handlerId).pipe(
        map(frame => frame.message as xtp.Message),
        takeUntil(onClose),
        takeUntil(onEnd),
        share(),
    )
    const watchMessageByType = getSubValues(onMessage, msg => msg.type)
    const onRemoteHandler = watchMessageByType("handlerInfo").pipe(
        map(msg => newRemoteHandler(ctx, msg.handlerInfo)),
        share(),
    )
    const onData = watchMessageByType('data').pipe(
        map(msg => msg.data),
        share(),
    )
    const availableDataSender = new Subject<Uint8Array>()
    const onSentData = availableDataSender.pipe(
        tap({
            next: data => frameSender.next({
                handlerId,
                yield: { data }
            }),
            complete: () => frameSender.next({
                handlerId,
                end: {}
            }),
            error: err => frameSender.next({
                handlerId,
                end: {
                    error: { typeName: err.name || 'Error', textMessage: err.message }
                }
            }),
        }),
        catchError(err => EMPTY),
        share(),
    )
    const handlerSender = new Subject<Handler>()
    const availableHandlerSender = new Subject<Handler>()
    const onSentHandler = availableHandlerSender.pipe(
        tap({
            next: handler => frameSender.next({
                handlerId,
                yield: { handlerInfo: handler.info },
            })
        }),
        share(),
    )
    const onAvailableAmount = merge(
        onSentData.pipe(mapTo(-1)),
        onSentHandler.pipe(
            mergeMap(it => it.theEnd),
            mapTo(-1),
        ),
        onMessage.pipe(mapTo(1)),
    ).pipe(
        scan((a, b) => a + b, 0),
        shareReplay(1),
    )
    const onAvailable = onAvailableAmount.pipe(
        map(amount => amount > 0),
        distinctUntilChanged(),
        shareReplay(1),
    )
    //side effect
    yieldedDataSender.pipe(
        withLatestFrom(onAvailable),
        tap({
            next: ([data, ok]) => {
                if (ok) availableDataSender.next(data)
            },
            complete: availableDataSender.complete,
            error: availableDataSender.error,
        }),
        catchError(err => EMPTY),
    ).subscribe()
    handlerSender.pipe(
        withLatestFrom(onAvailable),
        tap({
            next: ([handler, ok]) => {
                if (ok) availableHandlerSender.next(handler)
            },
        }),
        catchError(err => EMPTY),
    ).subscribe()
    return {
        info: handlerInfo,
        onData,
        onRemoteHandler,
        yieldedDataSender,
        theEnd: onEnd,
        yieldHandler: it=>handlerSender.next(it),
    }
}
